package com.Reactor3参考文档;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.TimeUnit;

import org.junit.Test;
import org.reactivestreams.Subscription;

import com.Reactor3参考文档.Push.MyEventListener;
import com.Reactor3参考文档.Push.MyEventProcessor;

import reactor.core.publisher.BaseSubscriber;
import reactor.core.publisher.Flux;

/**不像 push，create 可以用于 push 或 pull 模式，因此适合桥接监听器的 的 API，因为事件消息会随时异步地到来。回调方法 onRequest 可以被注册到 FluxSink 以便跟踪请求。这个回调可以被用于从源头请
 * 求更多数据，或者通过在下游请求到来 的时候传递数据给 sink 以实现背压管理。这是一种推送/拉取混合的模式， 因为下游可以从上游拉取已经就绪的数据，上游也可以在数据就绪的时候将其推送到下游。
 * @author zhangsulei
 *
 */
public class PushPull {

	public static interface MyMessageListener<T> {
		void onMessage(List<String> messages);
	}

	public static class MyMessageProcessor {

		private MyMessageListener<String> myMessageListener;

		public void register(MyMessageListener<String> myMessageListener) {
			this.myMessageListener = myMessageListener;
		}

		public List<String> request(long n) {
			List<String> l = new ArrayList<>();
			for(int i=0; i<n; i++){
				l.add("num" + i);
			}
			return l;
		}

	}

	@Test
	public void test(){
		MyMessageProcessor myMessageProcessor = new MyMessageProcessor();

		Flux<String> bridge = Flux.create(sink -> {
			// 后续异步到达的 message 也会被发送给 sink
			sink.onRequest(n -> {
				List<String> messages = myMessageProcessor.request(n);
				for (String s : messages) {
					sink.next(s);
				}
			})
			// onDispose 和 onCancel 这两个回调用于在被取消和终止后进行清理工作。 onDispose 可用于在 Flux 完成，有错误出现或被取消的时候执行清理。 onCancel 只用于针对“取消”信号执行相关操作，会先于 onDispose 执行。
			// onCancel 在取消时被调用。
			.onCancel(() -> {})
			// onDispose 在有完成、错误和取消时被调用
	        .onDispose(() -> {});
		});

		bridge.subscribe(new BaseSubscriber<String>() {  // 通过重写BaseSubscriber的方法来自定义Subscriber
            @Override
            protected void hookOnSubscribe(Subscription subscription) { // hookOnSubscribe定义在订阅的时候执行的操作
                System.out.println("Subscribed and make a request...");
                request(3); // 订阅时首先向上游请求3个元素
            }

            @Override
            protected void hookOnNext(String value) {  // hookOnNext定义每次在收到一个元素的时候的操作
                try {
                    TimeUnit.SECONDS.sleep(1);  // sleep 1秒钟来模拟慢的Subscriber
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println("Get value [" + value + "]");    // 打印收到的元素
                request(3); // 每次处理完1个元素后再请求3个（这里应该有一个计数，每次处理完3个再获取3个）
            }
        });
	}

}
